package org.ewancle.springboothivehdfs.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.sql.*;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
@Service
public class HiveService {

    private final DataSource dataSource;
    private final ExecutorService executorService;

    @Autowired
    public HiveService(DataSource hiveDataSource) {
        this.dataSource = hiveDataSource;
        // 创建线程池优化并发操作
        this.executorService = Executors.newFixedThreadPool(5);
    }

    /**
     * 执行查询SQL
     * @param sql 查询语句
     * @return 结果集列表，每行为一个Map
     */
    public List<Map<String, Object>> executeQuery(String sql) throws SQLException {
        log.info("Executing Hive query: {}", sql);
        List<Map<String, Object>> resultList = new ArrayList<>();

        try (Connection connection = dataSource.getConnection();
             Statement statement = connection.createStatement();
             ResultSet resultSet = statement.executeQuery(sql)) {

            ResultSetMetaData metaData = resultSet.getMetaData();
            int columnCount = metaData.getColumnCount();

            while (resultSet.next()) {
                Map<String, Object> row = new HashMap<>();
                for (int i = 1; i <= columnCount; i++) {
                    String columnName = metaData.getColumnName(i);
                    Object value = resultSet.getObject(i);
                    row.put(columnName, value);
                }
                resultList.add(row);
            }
        }

        return resultList;
    }

    /**
     * 异步执行查询
     * @param sql 查询语句
     * @return CompletableFuture对象
     */
    public CompletableFuture<List<Map<String, Object>>> executeQueryAsync(String sql) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                return executeQuery(sql);
            } catch (SQLException e) {
                log.error("Async Hive query failed: {}", sql, e);
                throw new RuntimeException("Hive query execution failed", e);
            }
        }, executorService);
    }

    /**
     * 执行更新SQL（DDL，DML等）
     * @param sql SQL语句
     * @return 影响的行数
     */
    public int executeUpdate(String sql) throws SQLException {
        log.info("Executing Hive update: {}", sql);
        try (Connection connection = dataSource.getConnection();
             Statement statement = connection.createStatement()) {
            return statement.executeUpdate(sql);
        }
    }

    /**
     * 检查表是否存在
     * @param database 数据库名
     * @param tableName 表名
     * @return 是否存在
     */
    public boolean tableExists(String database, String tableName) throws SQLException {
        String sql = "SHOW TABLES IN " + database + " LIKE '" + tableName + "'";
        List<Map<String, Object>> result = executeQuery(sql);
        return !result.isEmpty();
    }

    /**
     * 创建数据库
     * @param database 数据库名
     * @param ifNotExists 是否添加IF NOT EXISTS子句
     * @return 是否成功
     */
    public boolean createDatabase(String database, boolean ifNotExists) {
        String sql = "CREATE DATABASE " + (ifNotExists ? "IF NOT EXISTS " : "") + database;
        try {
            executeUpdate(sql);
            return true;
        } catch (SQLException e) {
            log.error("Create database failed: {}", database, e);
            return false;
        }
    }

    /**
     * 创建表
     * @param createTableSQL 建表SQL
     * @return 是否成功
     */
    public boolean createTable(String createTableSQL) {
        try {
            executeUpdate(createTableSQL);
            return true;
        } catch (SQLException e) {
            log.error("Create table failed: {}", createTableSQL, e);
            return false;
        }
    }

    /**
     * 删除表
     * @param database 数据库名
     * @param tableName 表名
     * @param ifExists 是否添加IF EXISTS子句
     * @return 是否成功
     */
    public boolean dropTable(String database, String tableName, boolean ifExists) {
        String sql = "DROP TABLE " + (ifExists ? "IF EXISTS " : "") + database + "." + tableName;
        try {
            executeUpdate(sql);
            return true;
        } catch (SQLException e) {
            log.error("Drop table failed: {}.{}", database, tableName, e);
            return false;
        }
    }

    /**
     * 获取表结构
     * @param database 数据库名
     * @param tableName 表名
     * @return 列信息列表
     */
    public List<Map<String, Object>> describeTable(String database, String tableName) throws SQLException {
        String sql = "DESCRIBE " + database + "." + tableName;
        return executeQuery(sql);
    }

    /**
     * 插入数据
     * @param database 数据库名
     * @param tableName 表名
     * @param data 列名和值的映射
     * @return 是否成功
     */
    public boolean insertData(String database, String tableName, Map<String, Object> data) {
        StringBuilder columnNames = new StringBuilder();
        StringBuilder values = new StringBuilder();

        boolean first = true;
        for (Map.Entry<String, Object> entry : data.entrySet()) {
            if (!first) {
                columnNames.append(", ");
                values.append(", ");
            }
            columnNames.append(entry.getKey());

            Object value = entry.getValue();
            if (value instanceof String) {
                values.append("'").append(value).append("'");
            } else {
                values.append(value);
            }

            first = false;
        }

        String sql = "INSERT INTO " + database + "." + tableName +
                " (" + columnNames + ") VALUES (" + values + ")";

        try {
            executeUpdate(sql);
            return true;
        } catch (SQLException e) {
            log.error("Insert data failed: {}.{}", database, tableName, e);
            return false;
        }
    }

    /**
     * 批量插入数据
     * @param database 数据库名
     * @param tableName 表名
     * @param dataList 数据列表
     * @return 是否成功
     */
    public boolean batchInsertData(String database, String tableName, List<Map<String, Object>> dataList) {
        if (dataList == null || dataList.isEmpty()) {
            return true;
        }

        // 处理列名
        Set<String> columnSet = new LinkedHashSet<>();
        for (Map<String, Object> data : dataList) {
            columnSet.addAll(data.keySet());
        }

        List<String> columnList = new ArrayList<>(columnSet);
        StringBuilder sb = new StringBuilder();
        sb.append("INSERT INTO ").append(database).append(".").append(tableName);
        sb.append(" (");

        for (int i = 0; i < columnList.size(); i++) {
            if (i > 0) sb.append(", ");
            sb.append(columnList.get(i));
        }

        sb.append(") VALUES ");

        // 处理值
        for (int i = 0; i < dataList.size(); i++) {
            Map<String, Object> data = dataList.get(i);
            if (i > 0) sb.append(", ");

            sb.append("(");
            for (int j = 0; j < columnList.size(); j++) {
                if (j > 0) sb.append(", ");

                Object value = data.get(columnList.get(j));
                if (value == null) {
                    sb.append("NULL");
                } else if (value instanceof String) {
                    sb.append("'").append(value).append("'");
                } else {
                    sb.append(value);
                }
            }
            sb.append(")");
        }

        try {
            executeUpdate(sb.toString());
            return true;
        } catch (SQLException e) {
            log.error("Batch insert data failed: {}.{}", database, tableName, e);
            return false;
        }
    }

/**
 * 更新数据
 * @param database 数据库名
 * @param tableName 表名
 * @param data 要更新的列名和值的映射
 * @param condition 条件语句，例如 "id = 1"
 * @return 是否成功
 */
public boolean updateData(String database, String tableName, Map<String, Object> data, String condition) {
    if (data == null || data.isEmpty()) {
        return false;
    }

    StringBuilder sb = new StringBuilder();
    sb.append("UPDATE ").append(database).append(".").append(tableName).append(" SET ");

    boolean first = true;
    for (Map.Entry<String, Object> entry : data.entrySet()) {
        if (!first) {
            sb.append(", ");
        }

        sb.append(entry.getKey()).append(" = ");
        Object value = entry.getValue();
        if (value == null) {
            sb.append("NULL");
        } else if (value instanceof String) {
            sb.append("'").append(value).append("'");
        } else {
            sb.append(value);
        }

        first = false;
    }

    if (condition != null && !condition.isEmpty()) {
        sb.append(" WHERE ").append(condition);
    }

    try {
        executeUpdate(sb.toString());
        return true;
    } catch (SQLException e) {
        log.error("Update data failed: {}.{}", database, tableName, e);
        return false;
    }
}

    /**
     * 删除数据
     * @param database 数据库名
     * @param tableName 表名
     * @param condition 条件语句，例如 "id = 1"
     * @return 是否成功
     */
    public boolean deleteData(String database, String tableName, String condition) {
        StringBuilder sb = new StringBuilder();
        sb.append("DELETE FROM ").append(database).append(".").append(tableName);

        if (condition != null && !condition.isEmpty()) {
            sb.append(" WHERE ").append(condition);
        }

        try {
            executeUpdate(sb.toString());
            return true;
        } catch (SQLException e) {
            log.error("Delete data failed: {}.{}", database, tableName, e);
            return false;
        }
    }

    /**
     * 执行HiveQL脚本
     * @param scriptPath 脚本路径
     * @return 是否成功
     */
    public boolean executeScript(String scriptPath) {
        try {
            String script = new String(java.nio.file.Files.readAllBytes(java.nio.file.Paths.get(scriptPath)));
            String[] statements = script.split(";");

            try (Connection connection = dataSource.getConnection();
                 Statement statement = connection.createStatement()) {

                for (String sql : statements) {
                    String trimmedSql = sql.trim();
                    if (!trimmedSql.isEmpty()) {
                        statement.execute(trimmedSql);
                    }
                }
            }
            return true;
        } catch (Exception e) {
            log.error("Execute script failed: {}", scriptPath, e);
            return false;
        }
    }

    /**
     * 关闭服务
     */
    public void close() {
        if (executorService != null) {
            executorService.shutdown();
        }
    }
}